package test

import java.util.concurrent.{Callable, TimeUnit, Future, ExecutorService}

/**
 * User: bigbully
 * Date: 15/2/21
 * Time: 下午5:36
 */
object Par {

  type Par[A] = ExecutorService => Future[A]

  def run[A](s: ExecutorService)(a: Par[A]): Future[A] = a(s)

  def unit[A](a: A): Par[A] = (es: ExecutorService) => UnitFuture(a) // `unit` is represented as a function that returns a `UnitFuture`, which is a simple implementation of `Future` that just wraps a constant value. It doesn't use the `ExecutorService` at all. It's always done and can't be cancelled. Its `get` method simply returns the value that we gave it.

  private case class UnitFuture[A](get: A) extends Future[A] {
    def isDone = true
    def get(timeout: Long, units: TimeUnit) = get
    def isCancelled = false
    def cancel(evenIfRunning: Boolean): Boolean = false
  }

  def map2[A,B,C](a: Par[A], b: Par[B])(f: (A,B) => C): Par[C] = {
    es=> {
      val fa = a(es)
      val fb = b(es)
      UnitFuture(f(fa.get(), fb.get()))
    }
  }

  def fork[A](a: => Par[A]): Par[A] = {
    es => {
      es.submit(new Callable[A] {
        override def call(): A = {
          a(es).get()
        }
      })
    }
  }

  def asyncF[A,B](f: A => B): A => Par[B] = {
    a => fork(unit(f(a)))
  }

  def map[A,B](fa: Par[A])(f: A => B): Par[B] = {
    es => UnitFuture(f(fa(es).get()))
  }

  def product[A,B](fa: Par[A], fb: Par[B]): Par[(A,B)] = {
    es => {
      UnitFuture((fa(es).get(), fb(es).get()))
    }
  }

  def map22[A,B,C](a: Par[A], b: Par[B])(f: (A,B) => C): Par[C] = {
    es => {
      val p = product(a, b)(es).get()
      UnitFuture(f(p._1, p._2))
    }
  }


  def sequenceRight[A](as: List[Par[A]]): Par[List[A]] =
    as match {
      case Nil => unit(Nil)
      case h :: t => map2(h, fork(sequence(t)))(_ :: _)
    }

  // We define `sequenceBalanced` using `IndexedSeq`, which provides an
  // efficient function for splitting the sequence in half.
  def sequenceBalanced[A](as: IndexedSeq[Par[A]]): Par[IndexedSeq[A]] = fork {
    if (as.isEmpty) unit(Vector())
    else if (as.length == 1) map(as.head)(a => Vector(a))
    else {
      val (l,r) = as.splitAt(as.length/2)
      map2(sequenceBalanced(l), sequenceBalanced(r))(_ ++ _)
    }
  }

  def sequence[A](as: List[Par[A]]): Par[List[A]] =
    map(sequenceBalanced(as.toIndexedSeq))(_.toList)

  def parFilter[A](l: List[A])(f: A => Boolean): Par[List[A]] = {
    val pars: List[Par[List[A]]] =
      l map (asyncF((a: A) => if (f(a)) List(a) else List()))
    map(sequence(pars))(_.flatten) // convenience method on `List` for concatenating a list of lists
  }

  def choice[A](cond: Par[Boolean])(t: Par[A], f: Par[A]): Par[A] = {
    es => if (run(es)(cond).get()) t(es) else f(es)
  }

  def choiceViaChoiceN[A](a: Par[Boolean])(ifTrue: Par[A], ifFalse: Par[A]): Par[A] = {
    choiceN(map(a)(b => if (b) 0 else 1))(List(ifTrue, ifFalse))
  }

  def choiceN[A](n: Par[Int])(choices: List[Par[A]]): Par[A] =
    es => {
      val ind = run(es)(n).get // Full source files
      run(es)(choices(ind))
    }


  def choiceViaFlatMap[A](p: Par[Boolean])(f: Par[A], t: Par[A]): Par[A] =
    flatMap(p)(b => if (b) t else f)

  /* `chooser` is usually called `flatMap` or `bind`. */
  def flatMap[A,B](p: Par[A])(choices: A => Par[B]): Par[B] =
    es => {
      val k = run(es)(p).get
      run(es)(choices(k))
    }

  def choiceMap[K,V](key: Par[K])(choices: Map[K,Par[V]]): Par[V] = {
    es => {
      val k = run(es)(key).get
      run(es)(choices(k))
    }
  }

  def chooser[A,B](p: Par[A])(choices: A => Par[B]): Par[B] = {
    es => {
      val k = run(es)(p).get
      run(es)(choices(k))
    }
  }


}
